package io.kiki.sba.registry.server.data_store.remoting.dataserver.handler;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.dataserver.DatumSummary;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffPublisherRequest;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffPublisherResult;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffUtils;
import io.kiki.sba.registry.common.model.slot.filter.SyncSlotAcceptorManager;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.slot.SlotManager;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;


public abstract class BaseSlotDiffPublisherRequestHandler extends AbstractServerHandler<DataSlotDiffPublisherRequest> {

    private static final Logger logger = LoggerFactory.getLogger(BaseSlotDiffPublisherRequestHandler.class);

    @Getter
    @Resource
    private DatumStorageDelegate datumStorageDelegate;

    @Autowired
    private ServerConfig serverConfig;

    @Getter
    @Autowired
    private SlotManager slotManager;

    public BaseSlotDiffPublisherRequestHandler() {

    }

    @Override
    public void checkParam(DataSlotDiffPublisherRequest request) {
        StringUtil.checkNonNegative(request.getSlotId(), "request.slotId");
        StringUtil.nullValue(request.getDatumSummaries(), "request.datumSummaries");
    }

    @Override
    public Object doHandle(Channel channel, DataSlotDiffPublisherRequest request) {
        try {
            slotManager.triggerUpdateSlots(request.getSlotsEpoch());
            final int slotId = request.getSlotId();
            if (!slotManager.isLeader(serverConfig.getDataCenter(), slotId)) {
                logger.warn("sync slot request from {}, not leader of {}", request.getLocalDataCenter(), slotId);
                return new GenericResponse().fillFailed("not leader of " + slotId);
            }

            Map<String, Map<String, Publisher>> existingPublishers = datumStorageDelegate.getDataInfoIdToRegisterIdToPublisherMapMap(serverConfig.getDataCenter(), request.getSlotId());

            DataSlotDiffPublisherResult result = calcDiffResult(request.getLocalDataCenter(), slotId, request.getDatumSummaries(), existingPublishers, request.getAcceptorManager());
            result.setSlotsEpoch(slotManager.getSlotsEpoch());
            return new GenericResponse().fillSucceed(result);
        } catch (Throwable e) {
            String msg = StringFormatter.format("DiffSyncPublisher request from {} error for slot {}", request.getLocalDataCenter(), request.getSlotId());
            logger.error(msg, e);
            return new GenericResponse().fillFailed(msg);
        }
    }

    private DataSlotDiffPublisherResult calcDiffResult(String requestDataCenter, int targetSlot, List<DatumSummary> newDatumSummaryList, Map<String, Map<String, Publisher>> oldPublishers, SyncSlotAcceptorManager syncSlotAcceptorManager) {
        DataSlotDiffPublisherResult result = DataSlotDiffUtils.diffPublishersResult(oldPublishers, newDatumSummaryList, serverConfig.getSlotSyncPublisherMaxNum(), syncSlotAcceptorManager);
        DataSlotDiffUtils.logDiffResult(requestDataCenter, result, targetSlot);
        return result;
    }

    @Override
    protected Node.NodeType getConnectNodeType() {
        return Node.NodeType.data_store;
    }

    @Override
    public Class interest() {
        return DataSlotDiffPublisherRequest.class;
    }

    @Override
    public Object buildFailedResponse(String msg) {
        return new GenericResponse().fillFailed(msg);
    }

}
